package com.ibm.cps.processors.window;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;

import org.apache.log4j.Logger;
import org.joda.time.DateTime;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.BasicProcessor;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

public class TimeWindowProcessor extends BasicProcessor{
	private static final Logger logger = Logger.getLogger(TimeWindowProcessor.class);

	private ITimeSeriesSchema parentSchema;
	private ITimeSeriesSchema outputSchema;
	private final Map<Comparable, TimeBasedBuffer> buffers = new HashMap<>();
	
	@ValueParam
	private int windowSizeInSecond;
	@ValueParam
	private int windowStepInSecond;
	
	public TimeWindowProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	@Override
	public void open() throws CPSException {
		this.outputSchema = createOutputSchema();
	}

	private ITimeSeriesSchema createOutputSchema() throws CPSException {
		ArrayNode arrayNode = JSONUtility.newArrayNode();
		byte[] types = parentSchema.getDataTypes();
		String[] names = parentSchema.getTagNames();
		for (int i = 2; i < types.length; i++) {
			String fieldName = names[i]+"_array";
			String fieldType = DataTypes.getDataType(types[i]);
			ObjectNode node = JSONUtility.newObjectNode();
			node.put(MessageFields.NAME, fieldName);
			if (fieldType.equals("int")) {
				node.put(MessageFields.DATATYPE, "int[]");
			}else if (fieldType.equals("double")) {
				node.put(MessageFields.DATATYPE, "double[]");
			}else if (fieldType.equals("string")) {
				node.put(MessageFields.DATATYPE, "string[]");
			}else {
				throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, "data type not support for array :" + fieldType);
			}
			arrayNode.add(node);
		}
		String schemaId = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaId, arrayNode, (new DateTime()).getMillis());
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		logger.info("======== Time Window input =========");
		logger.info(Arrays.toString(message.getObjectValues()));
		Comparable tsKey = message.getTsKey();
		TimeBasedBuffer buffer = buffers.get(tsKey);
		if(buffer == null){
			buffer = new TimeBasedBuffer(windowSizeInSecond);
			buffers.put(tsKey, buffer);
		}
		buffer.push(message);
		if(buffer.isFull()){
			Queue<ParsedDataMessage> bufferedData = buffer.getBufferedData();
			long oldestTimestamp = bufferedData.peek().getTimestamp();
			ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(outputSchema, tsKey, oldestTimestamp);
			for (int i = 2; i < outputSchema.getTagCount(); i++) {
				String originalFieldName = outputSchema.getTagNames()[i];
				Object[] tempFieldValue = new Object[bufferedData.size()];
				int j = 0;
				for (ParsedDataMessage insideWindow : bufferedData) {
					Object value = insideWindow.getValue(i);
					tempFieldValue[j] = value;
					j++;
				}
				builder = builder.append(originalFieldName, tempFieldValue);
			}
			List<ParsedDataMessage> result = new ArrayList<>();
			ParsedDataMessage parsedData = builder.toParsedDataMessage();
			result.add(parsedData);
			logger.info("====== Time Window Output ======");
			logger.info(Arrays.toString(outputSchema.getTagNames()));
			logger.info(Arrays.toString(parsedData.getObjectValues()));
			buffer.moveForward(windowStepInSecond);
			return result;
		}
		printBufferSize();
		return null;
	}
	
	private void printBufferSize(){
		logger.info("======= Time Window Buffer Size ======");
		for (Comparable tsKey : buffers.keySet()) {
			logger.info("======= " + tsKey + ":" + buffers.get(tsKey).size() + "======");
		}
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		return outputSchema;
	}

	@Override
	public void dispose() {
		
	}

}
